package com.whz.bus.configuration;

import com.whz.bus.consumer.DefaultChangeRecordConsumer;
import com.whz.bus.debezium.DebeziumEngineFactory;
import com.whz.bus.debezium.DebeziumProperties;
import com.whz.bus.debezium.DebeziumServerBootstrap;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

@Slf4j
@Configuration
@ConditionalOnClass({ DataSourceProperties.class })
@EnableConfigurationProperties({ DebeziumProperties.class, DataSourceProperties.class })
@ConditionalOnProperty(name = "tongyu.debezium.subscriber.enable", havingValue = "true", matchIfMissing = true)
public class DebeziumInitConfiguration {

    /**
     * db变化数据的消费者
     */
    @Bean("defaultChangeRecordConsumer")
    @ConditionalOnMissingBean
    public DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>> createChangeRecordConsumer() {
        return new DefaultChangeRecordConsumer();
    }

    /**
     * 创建debezium引擎
     */
    @Bean(value = "defaultDebeziumEngine", destroyMethod = "close")
    public DebeziumEngine<RecordChangeEvent<SourceRecord>> createDefaultEngine(DebeziumProperties debeziumProperties,
                                                                               DataSourceProperties dataSourceProperties) {
        DebeziumEngineFactory factory = DebeziumEngineFactory.builder()
                                                             .debeziumProperties(debeziumProperties)
                                                             .dataSourceProperties(dataSourceProperties)
                                                             .build();
        return factory.createKafkaConnectApiEngine(createChangeRecordConsumer());
    }

    /**
     * 订阅数据库变化启动类
     */
    @Bean
    @ConditionalOnMissingBean
    public DebeziumServerBootstrap<RecordChangeEvent<?>> createDebeziumServerBootstrap(
            List<DebeziumEngine<RecordChangeEvent<?>>> engines) {
        return new DebeziumServerBootstrap<>(engines);
    }
}